#include "util.h"
#include "utilc.h"
#include "easylog.h"
#include "easymqproducer.h"
#include <sstream>

using namespace util;

static void run_mq_thread(void* arg) {
    CEasyMqProducer* mq = (CEasyMqProducer*)(arg);
    mq->run();
}

CEasyMqProducer::CEasyMqProducer(void)
    : _Mq(NULL)
    , _multi(false)
    , _maxSize(0)
{
    uv_mutex_init(&_msglist_mutex);
    uv_cond_init(&_msglist_cond);

    uv_thread_t tid;
    uv_thread_create(&tid, run_mq_thread, this);
}

CEasyMqProducer::~CEasyMqProducer(void)
{
}

void CEasyMqProducer::load(std::string strName, bool bMulti/*=false*/) {
    m_strName   = strName;
    _multi      = bMulti;
    string host = Settings::getValue(strName,"host");
    int port    = Settings::getValue(strName,"port", 5672);
    string usr  = Settings::getValue(strName,"usr");
    string pwd  = Settings::getValue(strName,"pwd");
    string exc  = Settings::getValue(strName,"exc", "");
    string type = Settings::getValue(strName,"type", "fanout");
    string rout = Settings::getValue(strName,"rout", "");
    _Mq = MqClient::CMQProducer::Create(strName, host, port, usr, pwd, exc, type, rout);
    Log::debug("mq producer %s --> %s %d %s %s %s", strName.c_str(), host.c_str(), port, usr.c_str(), pwd.c_str(), exc.c_str());
}

void CEasyMqProducer::run() 
{
    std::list<std::string> tmp;
    while (true) {
        uv_mutex_lock(&_msglist_mutex);
        uv_cond_wait(&_msglist_cond, &_msglist_mutex);
        if(!_multi && _msglist.size() > 50) {
            Log::warning("mq msg %s %d", m_strName.c_str(), _msglist.size());
        } 
        if(_msglist.size() > 0) {
            Log::debug("mq msg %s %d", m_strName.c_str(), _msglist.size());
            for(auto &msg : _msglist) {
                tmp.push_back(msg);
            }
            _msglist.clear();
        }
        uv_mutex_unlock(&_msglist_mutex);

        if(_multi){
            while(tmp.size() > 0) {
                std::list<std::string> once_msg;
                if(_maxSize > 0 && tmp.size() > _maxSize) {
                    for(int i=0; i<_maxSize; ++i) {
                        once_msg.push_back(tmp.front());
                        tmp.pop_front();
                    }
                } else {
                    once_msg = tmp;
                    tmp.clear();
                }

                stringstream ss;
                bool first = true;
                ss << "[";
                for(auto &msg:once_msg) {
                    if(first) {
                        first = false;
                    } else {
                        ss << ",";
                    }
                    ss << msg;
                }
                ss << "]";
                if(_Mq)
                    _Mq->Send(ss.str());
            }
        } else {
            for(auto &msg : tmp) {
                if(_Mq)
                    _Mq->Send(msg);
            }
        }
        tmp.clear();
        sleep(1);
    }
}

void CEasyMqProducer::produce(std::string strMsg)
{
    uv_mutex_lock(&_msglist_mutex);
    _msglist.push_back(strMsg);
    uv_cond_signal(&_msglist_cond);
    uv_mutex_unlock(&_msglist_mutex);
}

void CEasyMqProducer::produce(std::list<std::string> lsMsg)
{
    uv_mutex_lock(&_msglist_mutex);
    _msglist.splice(_msglist.end(), lsMsg);
    uv_cond_signal(&_msglist_cond);
    uv_mutex_unlock(&_msglist_mutex);
}